//
//  IpcPort.cpp
//  2006, John Wiggins
//
// This program is free software; you can redistribute it and/or modify  
// it under the terms of the GNU General Public License as published by  
// the Free Software Foundation; either version 2 of the License, or     
// (at your option) any later version.                                   
// 
// This program is distributed in the hope that it will be useful,       
// but WITHOUT ANY WARRANTY; without even the implied warranty of        
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         
// GNU General Public License for more details.                          
//     
//     You should have received a copy of the GNU General Public License     
//     along with this program; if not, write to the                         
//         Free Software Foundation, Inc.,                                       
//         59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.  
//-------------------------------------------------------------
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <errno.h>
#include <unistd.h>
#include <assert.h>

#include "Ipc/IpcPort.h"
#include "Basics/BasErrors.h"
//-------------------------------------------------------------
typedef struct pp_msg
{
    uint32 code;
    int32 size;
    void *data;
    struct pp_msg *next;
} pp_msg_t;
  
typedef struct pp_port_data
{
    ipc_port_t id;
    char name[IPC_PORT_NAME_LENGTH];
    pthread_mutex_t lock;
    pthread_cond_t signal;
    pp_msg_t *head;
    pp_msg_t *tail;

    bool deletion_pending;
} pp_port_data_t;

typedef struct pp_port_list_item
{
    pp_port_data *item;
    pp_port_list_item *next;
} pp_port_list_item_t;

static bool _pp_is_initialized = false;
static pp_port_list_item *_pp_ports = NULL;
//static pthread_mutex_t _pp_list_lock;
pthread_mutex_t _pp_list_lock;
static ipc_port_t _pp_port_id_base = 1;
//-------------------------------------------------------------
static void
_pp_init()
{
    if ( !_pp_is_initialized )
    {
        /* create a global lock for serialized access to the port list */
        pthread_mutex_init( &_pp_list_lock, NULL );
    
        /* be extra sure of these values */
        _pp_ports = NULL;
        _pp_port_id_base = 1;
    
        /* that's it */
        _pp_is_initialized = true;
    }
}
//-------------------------------------------------------------
ipc_port_t
pp_create_port( char const *name )
{
    ipc_port_t ret = GE_NO_MEMORY;
    pp_port_data_t *nItem;
    pp_port_list_item_t *curr;
    
    /* make sure we're good to go */
    if ( !_pp_is_initialized )
    {
        _pp_init();
    }
    
    /* check to see if there is a port that already has this name */
    if ( pp_find_port( name ) > GE_NO_ERROR )
    {
        ret = GE_NAME_COLLISION;
    }
    else
    {
        /* initialize the new port */
        nItem = (pp_port_data *)malloc( sizeof( pp_port_data_t ) );
        curr = (pp_port_list_item *)malloc( sizeof( pp_port_list_item_t ) );
    
        if ( nItem != NULL && curr != NULL )
        {
            /* fill in data */
            nItem->deletion_pending = false;
            strncpy( nItem->name, name, IPC_PORT_NAME_LENGTH-1 );
            nItem->name[IPC_PORT_NAME_LENGTH] = 0;
            pthread_mutex_init( &(nItem->lock), NULL );
            pthread_cond_init( &(nItem->signal), NULL );
            nItem->head = nItem->tail = NULL;
        
            curr->item = nItem;
            
            /* lock the list mutex */
            pthread_mutex_lock( &_pp_list_lock );
            
            /* acquire a port id */
            /* no check to see if this id is already in use. This would only happen
            * in extreme cases (ie- assloads of ports created, most destroyed, then
            * the newly created port just happens to share the same id because
            * _pp_port_id_base overflowed). */
            ret = nItem->id = _pp_port_id_base++;
            /* add this port to the list */
            curr->next = _pp_ports;
            _pp_ports = curr;
        
            /* unlock the list mutex */
            pthread_mutex_unlock( &_pp_list_lock );
        }
        else
        {
            if ( nItem != NULL )
            {
                free( nItem );
            }
            if ( curr != NULL )
            {
                free( curr );
            }
        }
    }
    
    return ret;
}
//-------------------------------------------------------------
status_t
pp_delete_port( ipc_port_t port )
{
    status_t ret = GE_BAD_PORT_ID;
    pp_port_list_item_t *prev, *curr;
    pp_port_data_t *port_data;
    pp_msg_t *data, *data_tmp;
    
    /* make sure we're good to go */
    if ( !_pp_is_initialized )
    {
        _pp_init();
    }
    
    /* lock the list mutex */
    pthread_mutex_lock( &_pp_list_lock );
    
    prev = _pp_ports;
    curr = _pp_ports;
    while ( curr != NULL )
    {
        if ( curr->item->id == port )
        {
            /* delete the port */
            /* first, we point to the lucky item */
            port_data = curr->item;
            /* lock its mutex, make note of the deletion and signal its cond_var */
            pthread_mutex_lock( &(port_data->lock) );
            port_data->deletion_pending = true;
            /* signal all threads waiting on this port */
            pthread_cond_broadcast( &(port_data->signal) );
            pthread_mutex_unlock( &(port_data->lock) );
            /* then we remove it from the list */
            if ( curr == _pp_ports )
            {
                /* special case where the port being deleted is the list head */
                _pp_ports = curr->next;
            }
            else
            {
                /* normal case */
                prev->next = curr->next;
            }
            /* now delete the port data */
            /* free the message queue first (tedious) */
            data = port_data->head;
            while ( data != NULL )
            {
                data_tmp = data;
                data = data->next;
        
                free( data_tmp->data );
                data_tmp->data = NULL;
                free( data_tmp );
                data_tmp = NULL;
            }
            /* then get rid of it's mutex and cond_var */
            /* destroy the mutex (loop until destruction succeeds) */
            /* (according to the man page for pthread_mutex_destroy,
            * the function will return EBUSY if another thread has
            * locked or referenced it) */
            while ( pthread_mutex_destroy( &(port_data->lock) ) == EBUSY )
            {
                usleep( 1000 );
            }
            /* ditto for the condvar */
            while ( pthread_cond_destroy( &(port_data->signal) ) == EBUSY )
            {
                usleep(1000);
            }
            /* then the rest of the port data */
            free( port_data );
            port_data = NULL;
            free( curr );
            curr = NULL;
        
            /* success! */
            ret = GE_NO_ERROR;
            break;
        }
    
        prev = curr;
        curr = curr->next;
    }
    
    /* unlock the list mutex */
    pthread_mutex_unlock( &_pp_list_lock );
    
    return ret;
}
//-------------------------------------------------------------
ipc_port_t
pp_find_port( char const *name )
{
    ipc_port_t ret = GE_NAME_NOT_FOUND;
    pp_port_list_item_t *curr;
    
    /* make sure we're good to go */
    if ( !_pp_is_initialized )
    {
        _pp_init();
    }
    
    /* lock the list mutex */
    pthread_mutex_lock( &_pp_list_lock );
    
    /* step through the list */
    curr = _pp_ports;
    while ( curr != NULL )
    {
        if ( strncmp( name, curr->item->name, IPC_PORT_NAME_LENGTH-1 ) == 0 )
        {
            /* found it! */
            ret = curr->item->id;
            break;
        }
        /* next! */
        curr = curr->next;
    }
    
    /* unlock the list mutex */
    pthread_mutex_unlock( &_pp_list_lock );
    
    return ret;
}
//-------------------------------------------------------------
status_t
pp_port_has_message( ipc_port_t port )
{
    status_t ret = GE_BAD_PORT_ID;
    pp_port_list_item_t *curr;
    bool list_is_locked = true;
    
    /* make sure we're good to go */
    if ( !_pp_is_initialized )
    {
        _pp_init();
    }
    
    /* lock the list mutex */
    pthread_mutex_lock( &_pp_list_lock );
    
    /* step through the port list */
    curr = _pp_ports;
    while ( curr != NULL )
    {
        /* look for the port */
        if ( curr->item->id == port )
        {
            /* unlock the list mutex now that we have the port's lock */
            pthread_mutex_unlock( &_pp_list_lock );
            list_is_locked = false;
            /* lock the port */
            pthread_mutex_lock( &(curr->item->lock) );
            /* check the message queue for a message */
            if ( curr->item->head != NULL )
            {
                /* there is a message. get it's size. */
                ret = curr->item->head->size;
                /* unlock the port and bail */
                pthread_mutex_unlock( &(curr->item->lock) );
                break;
            }
            else
            {
                /* empty queue. return GE_NO_ERROR. */ 
                ret = GE_NO_ERROR;
                pthread_mutex_unlock( &(curr->item->lock) );
                break;
            }
        }
        curr = curr->next;
    }
    
    /* unlock the list mutex, only if it's still locked */
    if ( list_is_locked )
    {
        pthread_mutex_unlock( &_pp_list_lock );
    }
    
    return ret;
}
//-------------------------------------------------------------
status_t
pp_port_buffer_size( ipc_port_t port )
{
    status_t ret = GE_BAD_PORT_ID;
    pp_port_list_item_t *curr;
    bool list_is_locked = true;
    
    /* make sure we're good to go */
    if ( !_pp_is_initialized )
    {
        _pp_init();
    }
    
    /* lock the list mutex */
    pthread_mutex_lock( &_pp_list_lock );
    
    /* step through the port list */
    curr = _pp_ports;
    while ( curr != NULL )
    {
        /* look for the port */
        if ( curr->item->id == port )
        {
            /* unlock the list mutex now that we have the port's lock */
            pthread_mutex_unlock( &_pp_list_lock );
            list_is_locked = false;
            /* lock the port */
            pthread_mutex_lock( &(curr->item->lock) );
            /* check the message queue for a message */
            if ( curr->item->head != NULL )
            {
                /* there is a message. get it's size. */
                ret = curr->item->head->size;
                /* unlock the port and bail */
                pthread_mutex_unlock( &(curr->item->lock) );
                break;
            }
            else
            {
                /* empty queue. we must block until a message arrives */
                pthread_cond_wait( &(curr->item->signal), &(curr->item->lock) );
                /* at this point the port either has a message or has been destroyed */
                if ( curr->item->deletion_pending )
                {
                    /* the port is gone */
                    /* unlock its mutex and bail */
                    pthread_mutex_unlock( &(curr->item->lock) );
                    break;
                }
                else
                {
                    /* there is a message. (fucking better be) */
                    //XXX: there appears to be a race condition here
                    assert( curr->item->head );
                    /* get the size, unlock and bail */
                    ret = curr->item->head->size;
                    pthread_mutex_unlock( &(curr->item->lock) );
                    break;
                }
            }
        }
        curr = curr->next;
    }
    
    /* unlock the list mutex, only if it's still locked */
    if ( list_is_locked )
    {
        pthread_mutex_unlock( &_pp_list_lock );
    }
    
    return ret;
}
//-------------------------------------------------------------
status_t
pp_read_port( ipc_port_t port, uint32 *msg_code,
                void *msg_buffer, int32 buffer_size )
{
    status_t ret = GE_BAD_PORT_ID;
    pp_port_list_item_t *curr;
    pp_msg_t *temp;
    bool list_is_locked = true;
    
    /* make sure we're good to go */
    if ( !_pp_is_initialized )
    {
        _pp_init();
    }
    
    /* lock the list mutex */
    pthread_mutex_lock( &_pp_list_lock );
    
    /* step through the list of ports */
    curr = _pp_ports;
    while ( curr != NULL )
    {
        /* look for the port */
        if ( curr->item->id == port )
        {
            /* unlock the list mutex now that we have the port's lock */
            pthread_mutex_unlock( &_pp_list_lock );
            list_is_locked = false;
            /* lock the port */
            pthread_mutex_lock( &(curr->item->lock) );
            /* check the message queue for a message */
            if ( curr->item->head != NULL )
            {
                /* make sure there's enough space in the buffer for the whole message */
                if ( buffer_size >= curr->item->head->size )
                {
                    /* there is a message. copy it's contents. */
                    memcpy( msg_buffer, curr->item->head->data, curr->item->head->size );
                    *msg_code = curr->item->head->code;
                    /* remove the message from the queue */
                    temp = curr->item->head;
                    curr->item->head = curr->item->head->next;
                    /* watch out for tail removal */
                    if ( temp->next == NULL )
                    {
                        curr->item->tail = NULL;
                    }
                    /* free the memory for the message */
                    free( temp->data );
                    free( temp );
                    /* all is well */
                    ret = GE_NO_ERROR;
                    /* unlock the port and bail */
                    pthread_mutex_unlock( &(curr->item->lock) );
                    break;
                }
                else
                {
                    /* the read buffer is not big enough! */
                    ret = GE_BUFFER_TOO_SMALL;
                    *msg_code = 0;
                    /* unlock the port and bail */
                    pthread_mutex_unlock( &(curr->item->lock) );
                    break;
                }
            }
            else
            {
                /* empty queue. we must block until a message arrives */
                pthread_cond_wait( &(curr->item->signal), &(curr->item->lock) );
                /* at this point the port either has a message or has been destroyed */
                if ( curr->item->deletion_pending )
                {
                    /* the port is gone */
                    /* unlock its mutex and bail */
                    pthread_mutex_unlock( &(curr->item->lock) );
                    break;
                }
                else
                {
                    /* there is a message. (... there had fucking better be) */
                    assert( curr->item->head );
                    /* make sure there's enough space in the buffer for the whole message */
                    if ( buffer_size >= curr->item->head->size )
                    {
                        /* copy it's contents. */
                        memcpy( msg_buffer, curr->item->head->data, curr->item->head->size );
                        *msg_code = curr->item->head->code;
                        /* remove the message from the queue */
                        temp = curr->item->head;
                        curr->item->head = curr->item->head->next;
                        /* watch out for tail removal */
                        if ( temp->next == NULL )
                        {
                            curr->item->tail = NULL;
                        }
                        /* free the memory for the message */
                        free( temp->data );
                        free( temp );
                        /* all is well */
                        ret = GE_NO_ERROR;
                        /* unlock the port and bail */
                        pthread_mutex_unlock( &(curr->item->lock) );
                        break;
                    }
                    else
                    {
                        /* the read buffer is not big enough! */
                        ret = GE_BUFFER_TOO_SMALL;
                        *msg_code = 0;
                        /* unlock the port and bail */
                        pthread_mutex_unlock( &(curr->item->lock) );
                        break;
                    }
                }
            }
        }
        
        curr = curr->next;
    }
    
    /* unlock the list mutex, only if it's still locked */
    if ( list_is_locked )
    {
        pthread_mutex_unlock( &_pp_list_lock );
    }
    
    return ret;
}
//-------------------------------------------------------------
status_t
pp_write_port( ipc_port_t port, uint32 msg_code,
                void *msg_buffer, int32 buffer_size )
{
    status_t ret = GE_BAD_PORT_ID;
    bool list_is_locked = true;
    pp_port_list_item_t *curr = NULL;
    pp_msg_t *msg = NULL;
    
    /* make sure we're good to go */
    if ( !_pp_is_initialized )
    {
        _pp_init();
    }
    
    /* lock the list mutex */
    pthread_mutex_lock( &_pp_list_lock );
    
    /* step through the list of ports */
    curr = _pp_ports;
    while ( curr != NULL )
    {
        /* look for the port */
        if ( curr->item->id == port )
        {
            /* unlock the list mutex now that we have the port's lock */
            pthread_mutex_unlock( &_pp_list_lock );
            list_is_locked = false;
            /* lock the port */
            pthread_mutex_lock( &(curr->item->lock) );
            /* create the message */
            msg = (pp_msg_t *)malloc( sizeof(pp_msg_t) );
            if ( msg == NULL )
            {
                ret = GE_NO_MEMORY;
                pthread_mutex_unlock( &(curr->item->lock) );
                break;
            }
            msg->code = msg_code;
            msg->size = buffer_size;
            msg->next = NULL;
            msg->data = malloc( buffer_size );
            if ( msg->data == NULL )
            {
                free( msg );
                /* out of memory. unlock and bail. */
                ret = GE_NO_MEMORY;
                pthread_mutex_unlock( &(curr->item->lock) );
                break;
            }
            memcpy( msg->data, msg_buffer, buffer_size );
            /* add the message to the queue */
            if ( curr->item->tail == NULL )
            {
                /* empty queue, head and tail are now this message */
                curr->item->head = msg;
                curr->item->tail = msg;
                /* signal all waiting threads */
                pthread_cond_broadcast( &(curr->item->signal) );
                /* all is well. unlock and bail */
                ret = GE_NO_ERROR;
                pthread_mutex_unlock( &(curr->item->lock) );
                break;
            }
            else
            {
                /* non-empty queue, tail is now this message */
                curr->item->tail->next = msg;
                curr->item->tail = msg;
                /* signal any waiting threads */
                pthread_cond_broadcast( &(curr->item->signal) );
                /* all is well. unlock and bail */
                ret = GE_NO_ERROR;
                pthread_mutex_unlock( &(curr->item->lock) );
                break;
            }
        }
    
        curr = curr->next;
    }
    
    /* unlock the list mutex, only if it's still locked */
    if ( list_is_locked )
    {
        pthread_mutex_unlock( &_pp_list_lock );
    }
    
    return ret;
}
//-------------------------------------------------------------
